package com.example.rocketmq.demo.producer;

import com.example.rocketmq.demo.config.RocketMQConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

/**
 * 高级消息生产者
 * 提供更多同步和异步发送消息的选项
 */
@Slf4j
@Component
public class AdvancedProducer {

    @Autowired
    @Qualifier("defaultMQProducer")
    private DefaultMQProducer producer;

    /**
     * 同步发送消息并设置超时时间
     *
     * @param topic 主题
     * @param tag 标签
     * @param keys 消息键
     * @param content 消息内容
     * @param timeout 超时时间(毫秒)
     * @return 发送结果
     */
    public SendResult sendSyncWithTimeout(String topic, String tag, String keys, String content, long timeout) throws Exception {
        Message message = new Message(
                topic,
                tag,
                keys,
                content.getBytes(StandardCharsets.UTF_8)
        );

        SendResult sendResult = producer.send(message, timeout);
        log.info("同步发送消息结果(带超时): {}, topic={}, tag={}, keys={}", sendResult, topic, tag, keys);
        return sendResult;
    }

    /**
     * 异步发送消息并返回CompletableFuture
     *
     * @param topic 主题
     * @param tag 标签
     * @param keys 消息键
     * @param content 消息内容
     * @return 包含发送结果的CompletableFuture
     */
    public CompletableFuture<SendResult> sendAsyncWithFuture(String topic, String tag, String keys, String content) {
        Message message = new Message(
                topic,
                tag,
                keys,
                content.getBytes(StandardCharsets.UTF_8)
        );

        CompletableFuture<SendResult> resultFuture = new CompletableFuture<>();

        try {
            producer.send(message, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.info("异步发送消息成功(带Future): {}, topic={}, tag={}, keys={}", sendResult, topic, tag, keys);
                    resultFuture.complete(sendResult);
                }

                @Override
                public void onException(Throwable e) {
                    log.error("异步发送消息失败(带Future): topic={}, tag={}, keys={}", topic, tag, keys, e);
                    resultFuture.completeExceptionally(e);
                }
            });
        } catch (Exception e) {
            resultFuture.completeExceptionally(e);
        }

        return resultFuture;
    }

    /**
     * 批量异步发送多条消息
     *
     * @param topic 主题
     * @param tag 标签
     * @param contents 多条消息内容
     * @return 是否全部发送成功
     */
    public boolean sendMultipleAsync(String topic, String tag, String... contents) {
        boolean allSuccess = true;

        for (int i = 0; i < contents.length; i++) {
            String content = contents[i];
            String keys = "key-" + i;

            Message message = new Message(
                    topic,
                    tag,
                    keys,
                    content.getBytes(StandardCharsets.UTF_8)
            );

            try {
                producer.send(message, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        log.info("批量异步发送消息成功: {}, topic={}, tag={}, keys={}", sendResult, topic, tag, keys);
                    }

                    @Override
                    public void onException(Throwable e) {
                        log.error("批量异步发送消息失败: topic={}, tag={}, keys={}", topic, tag, keys, e);
                    }
                });
            } catch (Exception e) {
                log.error("批量异步发送消息异常: topic={}, tag={}, keys={}", topic, tag, keys, e);
                allSuccess = false;
            }
        }

        return allSuccess;
    }

    /**
     * 同步发送消息并等待消费者确认
     *
     * @param content 消息内容
     * @param timeout 超时时间(秒)
     * @return 发送结果
     */
    public String sendAndAwaitConsumption(String content, int timeout) throws Exception {
        // 生成唯一消息ID
        String messageId = "msg-" + System.currentTimeMillis() + "-" + Math.round(Math.random() * 100000);

        Message message = new Message(
                RocketMQConstant.TOPIC_NORMAL,
                RocketMQConstant.TAG_NORMAL,
                messageId,
                content.getBytes(StandardCharsets.UTF_8)
        );

        // 添加特殊属性，标记需要确认消费
        message.putUserProperty("needConfirm", "true");

        // 发送消息
        SendResult sendResult = producer.send(message);
        log.info("发送需确认消费的消息: {}, messageId={}", sendResult, messageId);

        // 模拟等待消费确认
        // 实际项目中可以使用Redis或其他方式实现消费确认机制
        log.info("等待消费确认，最多等待{}秒", timeout);
        TimeUnit.SECONDS.sleep(1);  // 模拟等待

        // 这里简化处理，假设消息已被消费
        return "消息已发送并假设已被消费，消息ID: " + messageId;
    }
}
